(self, sock, incoming, outgoing, func, *args, **kwargs)
| 2120 | self.assertIs(ss._sslobj.context, ctx2) |
| 2121 | |
| 2122 | def ssl_io_loop(self, sock, incoming, outgoing, func, *args, **kwargs): |
| 2123 | # A simple IO loop. Call func(*args) depending on the error we get |
| 2124 | # (WANT_READ or WANT_WRITE) move data between the socket and the BIOs. |
| 2125 | timeout = kwargs.get('timeout', support.SHORT_TIMEOUT) |
| 2126 | count = 0 |
| 2127 | for _ in support.busy_retry(timeout): |
| 2128 | errno = None |
| 2129 | count += 1 |
| 2130 | try: |
| 2131 | ret = func(*args) |
| 2132 | except ssl.SSLError as e: |
| 2133 | if e.errno not in (ssl.SSL_ERROR_WANT_READ, |
| 2134 | ssl.SSL_ERROR_WANT_WRITE): |
| 2135 | raise |
| 2136 | errno = e.errno |
| 2137 | # Get any data from the outgoing BIO irrespective of any error, and |
| 2138 | # send it to the socket. |
| 2139 | buf = outgoing.read() |
| 2140 | sock.sendall(buf) |
| 2141 | # If there's no error, we're done. For WANT_READ, we need to get |
| 2142 | # data from the socket and put it in the incoming BIO. |
| 2143 | if errno is None: |
| 2144 | break |
| 2145 | elif errno == ssl.SSL_ERROR_WANT_READ: |
| 2146 | buf = sock.recv(32768) |
| 2147 | if buf: |
| 2148 | incoming.write(buf) |
| 2149 | else: |
| 2150 | incoming.write_eof() |
| 2151 | if support.verbose: |
| 2152 | sys.stdout.write("Needed %d calls to complete %s().\n" |
| 2153 | % (count, func.__name__)) |
| 2154 | return ret |
| 2155 | |
| 2156 | def test_bio_handshake(self): |
| 2157 | sock = socket.socket(socket.AF_INET) |
no test coverage detected