diff options
author | Ben Banfield-Zanin <benbz@matrix.org> | 2021-02-16 13:33:20 +0000 |
---|---|---|
committer | Ben Banfield-Zanin <benbz@matrix.org> | 2021-02-16 13:33:20 +0000 |
commit | dcf1b9c276e22bb6f5200fc029301c4d40e87a1f (patch) | |
tree | 1f5badce24645d99534133a7a989069906088fff /tests/test_utils/__init__.py | |
parent | Merge remote-tracking branch 'origin/release-v1.24.0' into bbz/info-mainline-... (diff) | |
parent | Fixup CHANGES (diff) | |
download | synapse-bbz/info-mainline-1.27.0.tar.xz |
Merge remote-tracking branch 'origin/release-v1.27.0' into bbz/info-mainline-1.27.0 github/bbz/info-mainline-1.27.0 bbz/info-mainline-1.27.0
Diffstat (limited to 'tests/test_utils/__init__.py')
-rw-r--r-- | tests/test_utils/__init__.py | 39 |
1 files changed, 39 insertions, 0 deletions
diff --git a/tests/test_utils/__init__.py b/tests/test_utils/__init__.py index d232b72264..43898d8142 100644 --- a/tests/test_utils/__init__.py +++ b/tests/test_utils/__init__.py @@ -22,6 +22,13 @@ import warnings from asyncio import Future from typing import Any, Awaitable, Callable, TypeVar +from mock import Mock + +import attr + +from twisted.python.failure import Failure +from twisted.web.client import ResponseDone + TV = TypeVar("TV") @@ -80,3 +87,35 @@ def setup_awaitable_errors() -> Callable[[], None]: sys.unraisablehook = unraisablehook # type: ignore return cleanup + + +def simple_async_mock(return_value=None, raises=None) -> Mock: + # AsyncMock is not available in python3.5, this mimics part of its behaviour + async def cb(*args, **kwargs): + if raises: + raise raises + return return_value + + return Mock(side_effect=cb) + + +@attr.s +class FakeResponse: + """A fake twisted.web.IResponse object + + there is a similar class at treq.test.test_response, but it lacks a `phrase` + attribute, and didn't support deliverBody until recently. + """ + + # HTTP response code + code = attr.ib(type=int) + + # HTTP response phrase (eg b'OK' for a 200) + phrase = attr.ib(type=bytes) + + # body of the response + body = attr.ib(type=bytes) + + def deliverBody(self, protocol): + protocol.dataReceived(self.body) + protocol.connectionLost(Failure(ResponseDone())) |