|
|
@@ -22,6 +22,11 @@ import warnings |
|
|
|
from asyncio import Future |
|
|
|
from typing import Any, Awaitable, Callable, TypeVar |
|
|
|
|
|
|
|
import attr |
|
|
|
|
|
|
|
from twisted.python.failure import Failure |
|
|
|
from twisted.web.client import ResponseDone |
|
|
|
|
|
|
|
TV = TypeVar("TV") |
|
|
|
|
|
|
|
|
|
|
@@ -80,3 +85,25 @@ def setup_awaitable_errors() -> Callable[[], None]: |
|
|
|
sys.unraisablehook = unraisablehook # type: ignore |
|
|
|
|
|
|
|
return cleanup |
|
|
|
|
|
|
|
|
|
|
|
@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())) |