(self)
| 176 | self.assertEqual([(bytearray(b'data'), 4)], list(out._buffer)) |
| 177 | |
| 178 | def test_parse_length_payload(self): |
| 179 | out = aiohttp.FlowControlDataQueue(self.stream) |
| 180 | buf = aiohttp.ParserBuffer() |
| 181 | p = protocol.HttpPayloadParser(None).parse_length_payload(out, buf, 4) |
| 182 | next(p) |
| 183 | p.send(b'da') |
| 184 | p.send(b't') |
| 185 | try: |
| 186 | p.send(b'aline') |
| 187 | except StopIteration: |
| 188 | pass |
| 189 | |
| 190 | self.assertEqual(3, len(out._buffer)) |
| 191 | self.assertEqual(b'data', b''.join(d for d, _ in out._buffer)) |
| 192 | self.assertEqual(b'line', bytes(buf)) |
| 193 | |
| 194 | def test_parse_length_payload_eof(self): |
| 195 | out = aiohttp.FlowControlDataQueue(self.stream) |
nothing calls this directly
no test coverage detected