(
name: str, info: typing.Mapping[str, typing.Any], verbose: bool = False
)
| 210 | |
| 211 | |
| 212 | def trace( |
| 213 | name: str, info: typing.Mapping[str, typing.Any], verbose: bool = False |
| 214 | ) -> None: |
| 215 | console = rich.console.Console() |
| 216 | if name == "connection.connect_tcp.started" and verbose: |
| 217 | host = info["host"] |
| 218 | console.print(f"* Connecting to {host!r}") |
| 219 | elif name == "connection.connect_tcp.complete" and verbose: |
| 220 | stream = info["return_value"] |
| 221 | server_addr = stream.get_extra_info("server_addr") |
| 222 | console.print(f"* Connected to {server_addr[0]!r} on port {server_addr[1]}") |
| 223 | elif name == "connection.start_tls.complete" and verbose: # pragma: no cover |
| 224 | stream = info["return_value"] |
| 225 | ssl_object = stream.get_extra_info("ssl_object") |
| 226 | version = ssl_object.version() |
| 227 | cipher = ssl_object.cipher() |
| 228 | server_cert = ssl_object.getpeercert() |
| 229 | alpn = ssl_object.selected_alpn_protocol() |
| 230 | console.print(f"* SSL established using {version!r} / {cipher[0]!r}") |
| 231 | console.print(f"* Selected ALPN protocol: {alpn!r}") |
| 232 | if server_cert: |
| 233 | console.print("* Server certificate:") |
| 234 | console.print(format_certificate(server_cert)) |
| 235 | elif name == "http11.send_request_headers.started" and verbose: |
| 236 | request = info["request"] |
| 237 | print_request_headers(request, http2=False) |
| 238 | elif name == "http2.send_request_headers.started" and verbose: # pragma: no cover |
| 239 | request = info["request"] |
| 240 | print_request_headers(request, http2=True) |
| 241 | elif name == "http11.receive_response_headers.complete": |
| 242 | http_version, status, reason_phrase, headers = info["return_value"] |
| 243 | print_response_headers(http_version, status, reason_phrase, headers) |
| 244 | elif name == "http2.receive_response_headers.complete": # pragma: no cover |
| 245 | status, headers = info["return_value"] |
| 246 | http_version = b"HTTP/2" |
| 247 | reason_phrase = None |
| 248 | print_response_headers(http_version, status, reason_phrase, headers) |
| 249 | |
| 250 | |
| 251 | def download_response(response: Response, download: typing.BinaryIO) -> None: |
nothing calls this directly
no test coverage detected