(self)
| 200 | self.assertRaises(aiohttp.EofStream, p.throw, aiohttp.EofStream) |
| 201 | |
| 202 | def test_parse_chunked_payload(self): |
| 203 | out = aiohttp.FlowControlDataQueue(self.stream) |
| 204 | buf = aiohttp.ParserBuffer() |
| 205 | p = protocol.HttpPayloadParser(None).parse_chunked_payload(out, buf) |
| 206 | next(p) |
| 207 | try: |
| 208 | p.send(b'4\r\ndata\r\n4\r\nline\r\n0\r\ntest\r\n') |
| 209 | except StopIteration: |
| 210 | pass |
| 211 | self.assertEqual(b'dataline', b''.join(d for d, _ in out._buffer)) |
| 212 | self.assertEqual(b'', bytes(buf)) |
| 213 | |
| 214 | def test_parse_chunked_payload_chunks(self): |
| 215 | out = aiohttp.FlowControlDataQueue(self.stream) |
nothing calls this directly
no test coverage detected