(sock)
| 150 | auth = requests.auth.HTTPDigestAuth('user', 'pass') |
| 151 | |
| 152 | def digest_response_handler(sock): |
| 153 | # Respond to initial GET with a challenge. |
| 154 | request_content = consume_socket_content(sock, timeout=0.5) |
| 155 | assert request_content.startswith(b"GET / HTTP/1.1") |
| 156 | sock.send(text_401) |
| 157 | |
| 158 | # Verify we receive an Authorization header in response, then redirect. |
| 159 | request_content = consume_socket_content(sock, timeout=0.5) |
| 160 | assert expected_digest in request_content |
| 161 | sock.send(text_302) |
| 162 | |
| 163 | # Verify Authorization isn't sent to the redirected host, |
| 164 | # then send another challenge. |
| 165 | request_content = consume_socket_content(sock, timeout=0.5) |
| 166 | assert b'Authorization:' not in request_content |
| 167 | sock.send(text_401) |
| 168 | |
| 169 | # Verify Authorization is sent correctly again, and return 200 OK. |
| 170 | request_content = consume_socket_content(sock, timeout=0.5) |
| 171 | assert expected_digest in request_content |
| 172 | sock.send(text_200) |
| 173 | |
| 174 | return request_content |
| 175 | |
| 176 | close_server = threading.Event() |
| 177 | server = Server(digest_response_handler, wait_to_close_event=close_server) |
nothing calls this directly
no test coverage detected