(reader, writer)
| 213 | clients = [] |
| 214 | |
| 215 | async def handle_client(reader, writer): |
| 216 | nonlocal CNT |
| 217 | |
| 218 | data = await reader.readexactly(len(A_DATA)) |
| 219 | self.assertEqual(data, A_DATA) |
| 220 | writer.write(b'OK') |
| 221 | |
| 222 | data = await reader.readexactly(len(B_DATA)) |
| 223 | self.assertEqual(data, B_DATA) |
| 224 | writer.writelines([b'SP', bytearray(b'A'), memoryview(b'M')]) |
| 225 | |
| 226 | await writer.drain() |
| 227 | writer.close() |
| 228 | |
| 229 | CNT += 1 |
| 230 | |
| 231 | async def test_client(addr): |
| 232 | fut = asyncio.Future() |
nothing calls this directly
no test coverage detected