Merge pull request #133 from SukiCZ/assert_called

Assert called

README.rst

@@ -66,6 +66,7 @@ Supported HTTP methods: **GET**, **POST**, **PUT**, **PATCH**, **DELETE** and **
         resp = loop.run_until_complete(session.get('http://example.com'))
 
         assert resp.status == 200
+        mocked.assert_called_once_with('http://example.com')
 
 
 for convenience use *payload* argument to mock out json response. Example below.
@@ -88,6 +89,7 @@ for convenience use *payload* argument to mock out json response. Example below.
             data = loop.run_until_complete(resp.json())
 
             assert dict(foo='bar') == data
+            m.assert_called_once_with('http://test.example.com')
 
 **aioresponses allows to mock out any HTTP headers**
 
@@ -112,6 +114,7 @@ for convenience use *payload* argument to mock out json response. Example below.
         # note that we pass 'connection' but get 'Connection' (capitalized)
         # under the neath `multidict` is used to work with HTTP headers
         assert resp.headers['Connection'] == 'keep-alive'
+        m.assert_called_once_with('http://example.com', method='POST')
 
 **allows to register different responses for the same url**
 

aioresponses/core.py

@@ -334,6 +334,113 @@ def add(self, url: 'Union[URL, str, Pattern]', method: str = hdrs.METH_GET,
             callback=callback,
         ))
 
+    def _format_call_signature(self, *args, **kwargs) -> str:
+        message = '%s(%%s)' % self.__class__.__name__ or 'mock'
+        formatted_args = ''
+        args_string = ', '.join([repr(arg) for arg in args])
+        kwargs_string = ', '.join([
+            '%s=%r' % (key, value) for key, value in kwargs.items()
+        ])
+        if args_string:
+            formatted_args = args_string
+        if kwargs_string:
+            if formatted_args:
+                formatted_args += ', '
+            formatted_args += kwargs_string
+
+        return message % formatted_args
+
+    def assert_not_called(self):
+        """assert that the mock was never called.
+        """
+        if len(self.requests) != 0:
+            msg = ("Expected '%s' to not have been called. Called %s times."
+                   % (self.__class__.__name__,
+                      len(self._responses)))
+            raise AssertionError(msg)
+
+    def assert_called(self):
+        """assert that the mock was called at least once.
+        """
+        if len(self.requests) == 0:
+            msg = ("Expected '%s' to have been called."
+                   % (self.__class__.__name__,))
+            raise AssertionError(msg)
+
+    def assert_called_once(self):
+        """assert that the mock was called only once.
+        """
+        call_count = len(self.requests)
+        if call_count == 1:
+            call_count = len(list(self.requests.values())[0])
+        if not call_count == 1:
+            msg = ("Expected '%s' to have been called once. Called %s times."
+                   % (self.__class__.__name__,
+                      call_count))
+
+            raise AssertionError(msg)
+
+    def assert_called_with(self, url: 'Union[URL, str, Pattern]',
+                           method: str = hdrs.METH_GET,
+                           *args: Any,
+                           **kwargs: Any):
+        """assert that the last call was made with the specified arguments.
+
+        Raises an AssertionError if the args and keyword args passed in are
+        different to the last call to the mock."""
+        url = normalize_url(merge_params(url, kwargs.get('params')))
+        method = method.upper()
+        key = (method, url)
+        try:
+            expected = self.requests[key][-1]
+        except KeyError:
+            expected_string = self._format_call_signature(
+                url, method=method, *args, **kwargs
+            )
+            raise AssertionError(
+                '%s call not found' % expected_string
+            )
+        actual = self._build_request_call(method, *args, **kwargs)
+        if not expected == actual:
+            expected_string = self._format_call_signature(
+                expected,
+            )
+            actual_string = self._format_call_signature(
+                actual
+            )
+            raise AssertionError(
+                '%s != %s' % (expected_string, actual_string)
+            )
+
+    def assert_any_call(self, url: 'Union[URL, str, Pattern]',
+                        method: str = hdrs.METH_GET,
+                        *args: Any,
+                        **kwargs: Any):
+        """assert the mock has been called with the specified arguments.
+        The assert passes if the mock has *ever* been called, unlike
+        `assert_called_with` and `assert_called_once_with` that only pass if
+        the call is the most recent one."""
+        url = normalize_url(merge_params(url, kwargs.get('params')))
+        method = method.upper()
+        key = (method, url)
+
+        try:
+            self.requests[key]
+        except KeyError:
+            expected_string = self._format_call_signature(
+                url, method=method, *args, **kwargs
+            )
+            raise AssertionError(
+                '%s call not found' % expected_string
+            )
+
+    def assert_called_once_with(self, *args: Any, **kwargs: Any):
+        """assert that the mock was called once with the specified arguments.
+        Raises an AssertionError if the args and keyword args passed in are
+        different to the only call to the mock."""
+        self.assert_called_once()
+        self.assert_called_with(*args, **kwargs)
+
     @staticmethod
     def is_exception(resp_or_exc: Union[ClientResponse, Exception]) -> bool:
         if inspect.isclass(resp_or_exc):
@@ -408,12 +515,8 @@ async def _request_mock(self, orig_self: ClientSession,
 
         key = (method, url)
         self.requests.setdefault(key, [])
-        try:
-            kwargs_copy = copy.deepcopy(kwargs)
-        except (TypeError, ValueError):
-            # Handle the fact that some values cannot be deep copied
-            kwargs_copy = kwargs
-        self.requests[key].append(RequestCall(args, kwargs_copy))
+        request_call = self._build_request_call(method, *args, **kwargs)
+        self.requests[key].append(request_call)
 
         response = await self.match(method, url, **kwargs)
 
@@ -437,3 +540,19 @@ async def _request_mock(self, orig_self: ClientSession,
             response.raise_for_status()
 
         return response
+
+    def _build_request_call(self, method: str = hdrs.METH_GET,
+                            *args: Any,
+                            allow_redirects: bool = True,
+                            **kwargs: Any):
+        """Return request call."""
+        kwargs.setdefault('allow_redirects', allow_redirects)
+        if method == 'POST':
+            kwargs.setdefault('data', None)
+
+        try:
+            kwargs_copy = copy.deepcopy(kwargs)
+        except (TypeError, ValueError):
+            # Handle the fact that some values cannot be deep copied
+            kwargs_copy = kwargs
+        return RequestCall(args, kwargs_copy)

requirements-dev.txt

@@ -9,4 +9,5 @@ pytest-cov==2.10.1
 pytest-html==2.1.1
 ddt==1.4.1
 typing
-asynctest==0.13.0
+asynctest==0.13.0
+yarl==1.7.2

tests/test_aioresponses.py

@@ -143,13 +143,69 @@ async def test_returned_instance_and_params_handling(self, m):
         )
         self.assertIsInstance(response, ClientResponse)
         self.assertEqual(response.status, 200)
+        self.assertEqual(len(m.requests), 2)
+        with self.assertRaises(AssertionError):
+            m.assert_called_once()
 
     @aioresponses()
     def test_method_dont_match(self, m):
         m.get(self.url)
         with self.assertRaises(ClientConnectionError):
             self.run_async(self.session.post(self.url))
 
+    @aioresponses()
+    def test_post_with_data(self, m: aioresponses):
+        body = {'foo': 'bar'}
+        payload = {'spam': 'eggs'}
+        user_agent = {'User-Agent': 'aioresponses'}
+        m.post(
+            self.url,
+            payload=payload,
+            headers=dict(connection='keep-alive'),
+            body=body,
+        )
+        response = self.run_async(
+            self.session.post(
+                self.url,
+                data=payload,
+                headers=user_agent
+            )
+        )
+        self.assertIsInstance(response, ClientResponse)
+        self.assertEqual(response.status, 200)
+        response_data = self.run_async(response.json())
+        self.assertEqual(response_data, payload)
+        m.assert_called_once_with(
+            self.url,
+            method='POST',
+            data=payload,
+            headers={'User-Agent': 'aioresponses'}
+        )
+        # Wrong data
+        with self.assertRaises(AssertionError):
+            m.assert_called_once_with(
+                self.url,
+                method='POST',
+                data=body,
+                headers={'User-Agent': 'aioresponses'}
+            )
+        # Wrong url
+        with self.assertRaises(AssertionError):
+            m.assert_called_once_with(
+                'http://httpbin.org/',
+                method='POST',
+                data=payload,
+                headers={'User-Agent': 'aioresponses'}
+            )
+        # Wrong headers
+        with self.assertRaises(AssertionError):
+            m.assert_called_once_with(
+                self.url,
+                method='POST',
+                data=payload,
+                headers={'User-Agent': 'aiorequest'}
+            )
+
     @aioresponses()
     async def test_streaming(self, m):
         m.get(self.url, body='Test')
@@ -492,6 +548,60 @@ async def callback(url, **kwargs):
         data = self.run_async(response.read())
         assert data == body
 
+    @aioresponses()
+    def test_assert_not_called(self, m: aioresponses):
+        m.get(self.url)
+        m.assert_not_called()
+        self.run_async(self.session.get(self.url))
+        with self.assertRaises(AssertionError):
+            m.assert_not_called()
+
+    @aioresponses()
+    def test_assert_called(self, m: aioresponses):
+        m.get(self.url)
+        with self.assertRaises(AssertionError):
+            m.assert_called()
+        self.run_async(self.session.get(self.url))
+
+        m.assert_called_once()
+        m.assert_called_once_with(self.url)
+        m.assert_called_with(self.url)
+        with self.assertRaises(AssertionError):
+            m.assert_not_called()
+
+        with self.assertRaises(AssertionError):
+            m.assert_called_with("http://foo.bar")
+
+    @aioresponses()
+    async def test_assert_called_twice(self, m: aioresponses):
+        m.get(self.url, repeat=True)
+        m.assert_not_called()
+        await self.session.get(self.url)
+        await self.session.get(self.url)
+        with self.assertRaises(AssertionError):
+            m.assert_called_once()
+
+    @aioresponses()
+    async def test_assert_any_call(self, m: aioresponses):
+        http_bin_url = "http://httpbin.org"
+        m.get(self.url)
+        m.get(http_bin_url)
+        await self.session.get(self.url)
+        response = await self.session.get(http_bin_url)
+        self.assertEqual(response.status, 200)
+        m.assert_any_call(self.url)
+        m.assert_any_call(http_bin_url)
+
+    @aioresponses()
+    async def test_assert_any_call_not_called(self, m: aioresponses):
+        http_bin_url = "http://httpbin.org"
+        m.get(self.url)
+        response = await self.session.get(self.url)
+        self.assertEqual(response.status, 200)
+        m.assert_any_call(self.url)
+        with self.assertRaises(AssertionError):
+            m.assert_any_call(http_bin_url)
+
     @aioresponses()
     async def test_exception_requests_are_tracked(self, mocked):
         kwargs = {"json": [42], "allow_redirects": True}