Testing aiohttp client with unittest.mock.patch

2024/10/14 5:15:32

I've written a simple HTTP client using aiohttp and I'm trying to test it by patching aiohttp.ClientSession and aiohttp.ClientResponse. However, it appears as though the unittest.mock.patch decorator is not respecting my asynchronous code. At a guess, I would say it's some kind of namespacing mismatch.

Here's a minimal example:

from aiohttp import ClientSessionasync def is_ok(url:str) -> bool:async with ClientSession() as session:async with session.request("GET", url) as response:return (response.status == 200)

I'm using an asynchronous decorator for testing, as described in this answer. So here's my attempted test:

import unittest
from unittest.mock import MagicMock, patchfrom aiohttp import ClientResponsefrom my.original.module import is_okclass TestClient(unittest.TestCase):@async_test@patch("my.original.module.ClientSession", spec=True)async def test_client(self, mock_client):mock_response = MagicMock(spec=ClientResponse)mock_response.status = 200async def _mock_request(*args, **kwargs):return mock_responsemock_client.request = mock_responsestatus = await is_ok("foo")self.assertTrue(status)

My is_ok coroutine works fine when it's used in, say, __main__, but when I run the test, it gives me an error that indicates that the session.request function has not been mocked per my patch call. (Specifically it says "Could not parse hostname from URL 'foo'", which it should if it weren't mocked.)

I am unable to escape this behaviour. I have tried:

  • Importing is_ok after the mocking is done.
  • Various combinations of assigning mocks to mock_client and mock_client.__aenter__, setting mock_client.request to MagicMock(return_value=mock_response), or using mock_client().request, etc.
  • Writing a mock ClientSession with specific __aenter__ and __aexit__ methods and using it in the new argument to patch.

None of these appear to make a difference. If I put assertions into is_ok to test that ClientSession is an instance of MagicMock, then these assertions fail when I run the test (as, again, they would when the code is not patched). That leads me to my namespacing mismatch theory: That is, the event loop is running in a different namespace to which patch is targeting.

Either that, or I'm doing something stupid!

Answer

Mocking ClientSession is discouraged.

Recommended way is creation fake server and sending real requests to it.

Take a look on aiohttp example.

https://en.xdnf.cn/q/69449.html

Related Q&A

GridsearchCV: cant pickle function error when trying to pass lambda in parameter

I have looked quite extensively on stackoverflow and elsewhere and I cant seem to find an answer to the problem below. I am trying to modify a parameter of a function that is itself a parameter inside …

How to insert a carriage return in a ReportLab paragraph?

Is there a way to insert a carriage return in a Paragraph in ReportLab? I am trying to concatenate a "\n" to my paragraph string but this isnt working. Title = Paragraph("Title" + …

How to get predictions and calculate accuracy for a given test set in fast ai?

Im trying to load a learner which was exported by learn.export() and I want to run it against a test set. I want my test set have labels so that I can measure its accuracy. This is my code: test_src = …

Splitting the legend in matploblib

Is it possible to split up a single big legend into multiple (usually 2) smaller ones.from pylab import *t = arange(0.0, 2.0, 0.01) s = sin(2*pi*t) plot(t, s, linewidth=1.0, label="Graph1") g…

Python 3.x - iloc throws error - single positional indexer is out-of-bounds

I am scraping election data from a website and trying to store it in a dataframe import pandas as pd import bs4 import requestscolumns = [Candidate,Party,Criminal Cases,Education,Age,Total Assets,Liabi…

Supposed automatically threaded scipy and numpy functions arent making use of multiple cores

I am running Mac OS X 10.6.8 and am using the Enthought Python Distribution. I want for numpy functions to take advantage of both my cores. I am having a problem similar to that of this post: multithre…

Golang net.Listen binds to port thats already in use

Port 8888 is already bound on my (OS X 10.13.5) system, by a process running inside a docker container:$ netstat -an | grep 8888 tcp6 0 0 ::1.8888 *.* LISTE…

Aiohttp, Asyncio: RuntimeError: Event loop is closed

I have two scripts, scraper.py and db_control.py. In scraper.py I have something like this: ... def scrape(category, field, pages, search, use_proxy, proxy_file):...loop = asyncio.get_event_loop()to_do…

Python for ios interpreter [duplicate]

This question already has answers here:Closed 11 years ago.Possible Duplicate:Python or Ruby Interpreter on iOS I just discovered this apps pypad and python for ios They have like an interpreter an ed…

Detect when multiprocessing queue is empty and closed

Lets say I have two processes: a reader and a writer. How does the writer detect when the reader has finished writing values?The multiprocessing module has a queue with a close method that seems custo…