(loop, *, listen_addr=('127.0.0.1', 0),
use_ssl=False, router=None)
| 32 | |
| 33 | @contextlib.contextmanager |
| 34 | def run_server(loop, *, listen_addr=('127.0.0.1', 0), |
| 35 | use_ssl=False, router=None): |
| 36 | properties = {} |
| 37 | transports = [] |
| 38 | |
| 39 | class HttpRequestHandler: |
| 40 | |
| 41 | def __init__(self, addr): |
| 42 | if isinstance(addr, tuple): |
| 43 | host, port = addr |
| 44 | self.host = host |
| 45 | self.port = port |
| 46 | else: |
| 47 | self.host = host = 'localhost' |
| 48 | self.port = port = 0 |
| 49 | self.address = addr |
| 50 | self._url = '{}://{}:{}'.format( |
| 51 | 'https' if use_ssl else 'http', host, port) |
| 52 | |
| 53 | def __getitem__(self, key): |
| 54 | return properties[key] |
| 55 | |
| 56 | def __setitem__(self, key, value): |
| 57 | properties[key] = value |
| 58 | |
| 59 | def url(self, *suffix): |
| 60 | return urllib.parse.urljoin( |
| 61 | self._url, '/'.join(str(s) for s in suffix)) |
| 62 | |
| 63 | class TestHttpServer(server.ServerHttpProtocol): |
| 64 | |
| 65 | def connection_made(self, transport): |
| 66 | transports.append(transport) |
| 67 | |
| 68 | super().connection_made(transport) |
| 69 | |
| 70 | def handle_request(self, message, payload): |
| 71 | if properties.get('close', False): |
| 72 | return |
| 73 | |
| 74 | for hdr, val in message.headers.items(): |
| 75 | if (hdr == 'EXPECT') and (val == '100-continue'): |
| 76 | self.transport.write(b'HTTP/1.0 100 Continue\r\n\r\n') |
| 77 | break |
| 78 | |
| 79 | if router is not None: |
| 80 | body = yield from payload.read() |
| 81 | |
| 82 | rob = router( |
| 83 | self, properties, self.transport, message, body) |
| 84 | rob.dispatch() |
| 85 | |
| 86 | else: |
| 87 | response = aiohttp.Response(self.writer, 200, message.version) |
| 88 | |
| 89 | text = b'Test message' |
| 90 | response.add_header('Content-type', 'text/plain') |
| 91 | response.add_header('Content-length', str(len(text))) |
nothing calls this directly
no test coverage detected