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