Mock aiohttp ClientSession
(
http_method: str = "get",
return_value: str | dict | list | None = None,
status: int = 200,
)
| 48 | |
| 49 | |
| 50 | def mock_http_client( |
| 51 | http_method: str = "get", |
| 52 | return_value: str | dict | list | None = None, |
| 53 | status: int = 200, |
| 54 | ): |
| 55 | """Mock aiohttp ClientSession""" |
| 56 | |
| 57 | if isinstance(return_value, (dict, list)): |
| 58 | response_text = json.dumps(return_value) |
| 59 | else: |
| 60 | response_text = return_value or "{}" |
| 61 | |
| 62 | mock_response = MagicMock() |
| 63 | mock_response.status = status |
| 64 | mock_response.text = AsyncMock(return_value=response_text) |
| 65 | mock_response.json = AsyncMock( |
| 66 | return_value=json.loads(response_text) if response_text else {} |
| 67 | ) |
| 68 | mock_response.read = AsyncMock(return_value=response_text.encode()) |
| 69 | mock_response.__aenter__ = AsyncMock(return_value=mock_response) |
| 70 | mock_response.__aexit__ = AsyncMock(return_value=None) |
| 71 | mock_session = MagicMock() |
| 72 | |
| 73 | async def mock_method(*args, **kwargs): |
| 74 | return mock_response |
| 75 | |
| 76 | setattr(mock_session, http_method.lower(), mock_method) |
| 77 | mock_session.__aenter__ = AsyncMock(return_value=mock_session) |
| 78 | mock_session.__aexit__ = AsyncMock(return_value=None) |
| 79 | |
| 80 | mock_session.close = AsyncMock() |
| 81 | |
| 82 | return mock_session |
| 83 | |
| 84 | |
| 85 | def mock_file_message(input: str): |