()
| 262 | main_thread = threading.get_ident() |
| 263 | |
| 264 | def run_client(): |
| 265 | http = HTTPConnection(*server.server_address) |
| 266 | http.request("GET", "/") |
| 267 | with http.getresponse() as response: |
| 268 | response.read(100) |
| 269 | # The main thread should now be blocking in a send() system |
| 270 | # call. But in theory, it could get interrupted by other |
| 271 | # signals, and then retried. So keep sending the signal in a |
| 272 | # loop, in case an earlier signal happens to be delivered at |
| 273 | # an inconvenient moment. |
| 274 | while True: |
| 275 | pthread_kill(main_thread, signal.SIGUSR1) |
| 276 | if interrupted.wait(timeout=float(1)): |
| 277 | break |
| 278 | nonlocal received |
| 279 | received = len(response.read()) |
| 280 | http.close() |
| 281 | |
| 282 | background = threading.Thread(target=run_client) |
| 283 | background.start() |
nothing calls this directly
no test coverage detected