(self)
| 290 | client.stop() |
| 291 | |
| 292 | def test_create_connection_ssl_1(self): |
| 293 | self.loop.set_exception_handler(None) |
| 294 | |
| 295 | CNT = 0 |
| 296 | TOTAL_CNT = 25 |
| 297 | |
| 298 | A_DATA = b'A' * 1024 * BUF_MULTIPLIER |
| 299 | B_DATA = b'B' * 1024 * BUF_MULTIPLIER |
| 300 | |
| 301 | sslctx = self._create_server_ssl_context( |
| 302 | test_utils.ONLYCERT, |
| 303 | test_utils.ONLYKEY |
| 304 | ) |
| 305 | client_sslctx = self._create_client_ssl_context() |
| 306 | |
| 307 | def server(sock): |
| 308 | sock.starttls( |
| 309 | sslctx, |
| 310 | server_side=True) |
| 311 | |
| 312 | data = sock.recv_all(len(A_DATA)) |
| 313 | self.assertEqual(data, A_DATA) |
| 314 | sock.send(b'OK') |
| 315 | |
| 316 | data = sock.recv_all(len(B_DATA)) |
| 317 | self.assertEqual(data, B_DATA) |
| 318 | sock.send(b'SPAM') |
| 319 | |
| 320 | sock.close() |
| 321 | |
| 322 | async def client(addr): |
| 323 | reader, writer = await asyncio.open_connection( |
| 324 | *addr, |
| 325 | ssl=client_sslctx, |
| 326 | server_hostname='', |
| 327 | ssl_handshake_timeout=HANDSHAKE_TIMEOUT) |
| 328 | |
| 329 | writer.write(A_DATA) |
| 330 | self.assertEqual(await reader.readexactly(2), b'OK') |
| 331 | |
| 332 | writer.write(B_DATA) |
| 333 | self.assertEqual(await reader.readexactly(4), b'SPAM') |
| 334 | |
| 335 | nonlocal CNT |
| 336 | CNT += 1 |
| 337 | |
| 338 | writer.close() |
| 339 | await self.wait_closed(writer) |
| 340 | |
| 341 | async def client_sock(addr): |
| 342 | sock = socket.socket() |
| 343 | sock.connect(addr) |
| 344 | reader, writer = await asyncio.open_connection( |
| 345 | sock=sock, |
| 346 | ssl=client_sslctx, |
| 347 | server_hostname='', |
| 348 | ssl_handshake_timeout=HANDSHAKE_TIMEOUT) |
| 349 |
nothing calls this directly
no test coverage detected